package cn.azzhu.day07

import cn.azzhu.day02.SensorSource
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, tableConversions}
import org.apache.flink.types.Row

/**
 * @author azzhu
 * @create 2020-09-23 22:10:54
 */
object SQLEventTime {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val settings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val tEnv = StreamTableEnvironment.create(env, settings)

    val stream = env.addSource(new SensorSource)
      .assignAscendingTimestamps(_.timestamp) //必须分配时间戳和设置水位线

    tEnv.createTemporaryView("sensor",stream,'id,'timestamp.rowtime as 'ts,'temperature)

    tEnv.sqlQuery("select id,count(id) from sensor group by id,hop(ts,interval '5' second,interval '10' second)")
      .toAppendStream[Row]
      .print()

    env.execute("EventTimeSQL")
  }
}
